package com.junjie.food.orderServiceManager.service.impl;

import MountainMq.sender.TransMessageSender;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.junjie.food.orderServiceManager.dao.OrderDetailDao;
import com.junjie.food.orderServiceManager.dto.OrderMessageDTO;
import com.junjie.food.orderServiceManager.enummeration.OrderStatus;
import com.junjie.food.orderServiceManager.pojo.OrderDetailPo;
import com.junjie.food.orderServiceManager.service.OrderService;
import com.junjie.food.orderServiceManager.vo.OrderCreateVo;
import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.Date;

// 处理订单业务逻辑
@Service
@Slf4j
public class OrderServiceImpl implements OrderService {
    @Autowired
    private OrderDetailDao orderDetailDao;
    @Autowired(required = false)
    private RabbitTemplate rabbitTemplate;

    // 注入分布式事务版MQ信息发送者
    @Autowired
    private TransMessageSender transMessageSender;

    ObjectMapper objectMapper = new ObjectMapper();

    /**
     * 使用RabbitTemplate发送消息
     * @param orderCreateVo
     * @throws Exception
     */
    @Override
    public void createOrder(OrderCreateVo orderCreateVo) throws Exception{
        // 保存订单
        log.info("createOrder:orderCreateVO:{}", orderCreateVo);
        OrderDetailPo orderPO = new OrderDetailPo();
        orderPO.setAddress(orderCreateVo.getAddress());
        orderPO.setAccountId(orderCreateVo.getAccountId());
        orderPO.setProductId(orderCreateVo.getProductId());
        orderPO.setStatus(OrderStatus.ORDER_CREATING);
        orderPO.setDate(new Date());
        orderDetailDao.insert(orderPO);

        // 创建message类
        OrderMessageDTO orderMessageDTO = new OrderMessageDTO();
        orderMessageDTO.setOrderId(orderPO.getId());
        orderMessageDTO.setProductId(orderPO.getProductId());
        orderMessageDTO.setAccountId(orderCreateVo.getAccountId());

        // 将消息转换成json
        String messageToSend = objectMapper.writeValueAsString(orderMessageDTO);
        // 消息属性(必须创建，否则发送消息报异常说properties不能为null)
        MessageProperties properties = new MessageProperties();
        // 设置ttl
        properties.setExpiration("15000");
        // 消息体
        Message message = new Message(messageToSend.getBytes(),properties);
        // 创建CorrelationData对象 用于消息确认回调方法中 打印该消息是谁
        CorrelationData correlationData = new CorrelationData();
        correlationData.setId(orderPO.getId().toString());
        // 发送消息 记得传入CorrelationData
//        rabbitTemplate.send("exchange.order.restaurant",
//                "key.restaurant",
//                message,
//                correlationData
//                );

        // 使用自研分布式事务版MQ发送信息
        transMessageSender.send(
                "exchange.order.restaurant",
                "key.restaurant",
                orderMessageDTO);


        // 使用convertAndSend 可自动将string封装成message对象
//        rabbitTemplate.convertAndSend("exchange.order.restaurant",
//                "key.restaurant",
//                messageToSend
//        );

        // 可以选择自己操作channel
//        rabbitTemplate.execute(channel -> {
//
//        });
    }

    /**
     * 单条同步确认机制 原始方式发送消息
     */
    public void createOrder2(OrderCreateVo orderCreateVo) throws Exception{
        // 保存订单
        log.info("createOrder:orderCreateVO:{}", orderCreateVo);
        OrderDetailPo orderPO = new OrderDetailPo();
        orderPO.setAddress(orderCreateVo.getAddress());
        orderPO.setAccountId(orderCreateVo.getAccountId());
        orderPO.setProductId(orderCreateVo.getProductId());
        orderPO.setStatus(OrderStatus.ORDER_CREATING);
        orderPO.setDate(new Date());
        orderDetailDao.insert(orderPO);

        // 创建message类
        OrderMessageDTO orderMessageDTO = new OrderMessageDTO();
        orderMessageDTO.setOrderId(orderPO.getId());
        orderMessageDTO.setProductId(orderPO.getProductId());
        orderMessageDTO.setAccountId(orderCreateVo.getAccountId());

        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("110.40.176.175");

        try (Connection connection = connectionFactory.newConnection();
             Channel channel = connection.createChannel()) {
            // 开启消息确认机制
            channel.confirmSelect();

            // 发送消息应该是字符串json，所以这里使用序列化操作
            String messageToSend = objectMapper.writeValueAsString(orderMessageDTO);
            // 指定交换机 route_key 消息特殊的参数 消息体本身(要用字节)
            // 通知商家，让商家去确认订单

            // 这里发送10条消息是为了模拟 多条同步消息不好的地方 发了十条消息，但是哪个消息失败了是看不出来的
//            for (int i = 0; i < 10; i++)
            channel.basicPublish("exchange.order.restaurant", "key.restaurant", null, messageToSend.getBytes());

            log.info("message sent");
            // 判断消息是否成功发出去
            if (channel.waitForConfirms()) {
                log.info("RabbitMQ confirm success");
            }else{
                log.info("RabbitMQ confirm failed");
            }
        }
    }

    /**
     * 异步确认机制
     */
    public void createOrder3(OrderCreateVo orderCreateVo) throws Exception{
        // 保存订单
        log.info("createOrder:orderCreateVO:{}", orderCreateVo);
        OrderDetailPo orderPO = new OrderDetailPo();
        orderPO.setAddress(orderCreateVo.getAddress());
        orderPO.setAccountId(orderCreateVo.getAccountId());
        orderPO.setProductId(orderCreateVo.getProductId());
        orderPO.setStatus(OrderStatus.ORDER_CREATING);
        orderPO.setDate(new Date());
        orderDetailDao.insert(orderPO);

        // 创建message类
        OrderMessageDTO orderMessageDTO = new OrderMessageDTO();
        orderMessageDTO.setOrderId(orderPO.getId());
        orderMessageDTO.setProductId(orderPO.getProductId());
        orderMessageDTO.setAccountId(orderCreateVo.getAccountId());

        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("110.40.176.175");

        try (Connection connection = connectionFactory.newConnection();
             Channel channel = connection.createChannel()) {
            // 开启消息确认机制
            channel.confirmSelect();
            // 异步消息确认机制，如果成功/失败，MQ会回调这里的方法
            ConfirmListener confirmListener = new ConfirmListener() {
                // deliveryTag是发送端的编号(发送的第几条消息，第一条编号就是1)
                // multiple是true就是多条消息，false就是单条消息
                @Override   // 确认成功调用该方法
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    log.info("Ack,deliveryTag:{},multiple:{}",deliveryTag,multiple);
                }

                @Override  // 确认失败调用该方法
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    log.info("Nack,deliveryTag:{},multiple:{}",deliveryTag,multiple);
                }
            };

            // 发送消息应该是字符串json，所以这里使用序列化操作
            String messageToSend = objectMapper.writeValueAsString(orderMessageDTO);
            // 指定交换机 route_key 消息特殊的参数 消息体本身(要用字节)
            // 通知商家，让商家去确认订单

            // 这个是AMQP里的内部类 这里面没有提供set方法，所以我们使用build建造者设置值
            // 这里消息过期时间设置成了15s
            AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("15000").build();

            // 这里发送50条消息是为了模拟 多条同步消息不好的地方 发了十条消息，但是哪个消息失败了是看不出来的
            // 同时这里还模拟了RabbitMQ消息限流机制
//            for (int i = 0; i < 50; i++){
                // 这里设置了 properties，主要是设置消息过期时间的 平时不用可以直接传递null
                channel.basicPublish("exchange.order.restaurant", "key.restaurant", properties, messageToSend.getBytes());
                log.info("message sent");
//            }
            // 要睡眠一下，否则上面的方法无法收到回调
            Thread.sleep(100000);
        }
    }

}
